package com.example.rxjavacoredemo.rxjava;

public class ObservableCreator<T> extends Observable<T> {

    final ObservableOnSubscribe<T> source;

    public ObservableCreator(ObservableOnSubscribe<T> source) {
        this.source = source;
    }


    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        CreatorEmitter<T> creatorEmitter = new CreatorEmitter<>(observer);
        observer.onSubscribe(creatorEmitter);
        try {
            source.subscribe(creatorEmitter);
        } catch (Exception e) {
            e.printStackTrace();
            creatorEmitter.onError(e);
        }
    }

    static final class CreatorEmitter<T> implements ObservableEmitter<T>, Disposable {

        final Observer<? super T> observer;
        private boolean bool;

        public CreatorEmitter(Observer<? super T> observer) {
            this.observer = observer;
        }

        @Override
        public void dispose(boolean bool) {
            this.bool = bool;
        }

        @Override
        public boolean isDisposable() {
            return bool;
        }

        @Override
        public void onNext(T value) {
            if (bool) {
                return;
            }
            observer.onNext(value);
        }

        @Override
        public void onError(Throwable throwable) {
            if (bool) {
                return;
            }
            observer.onError(throwable);
        }

        @Override
        public void onComplete() {
            if (bool) {
                return;
            }
            observer.onComplete();
        }
    }

}
